package com.ds.lens.flink.monitor.metricsjob;

import lombok.extern.slf4j.Slf4j;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.util.CollectionUtil;
import org.apache.flink.util.Collector;

import java.util.List;
import java.util.Map;

@Slf4j
public class MetricsProcessFunction extends ProcessFunction<Metrics, MetricsResult> {
    @Override
    public void processElement(Metrics value, Context ctx, Collector<MetricsResult> out) {
        try {
            List<Metric> metrics = value.getMetrics();
            if (!CollectionUtil.isNullOrEmpty(metrics)) {
                for (Metric m : metrics) {
                    List<List<Double>> points = m.getPoints();
                    if (!CollectionUtil.isNullOrEmpty(points)) {
                        for (List<Double> p : points) {

                            if (!CollectionUtil.isNullOrEmpty(p) && p.size() == 2) {
                                Long timeStamp = (p.get(0)).longValue();
                                Float metricValue =  Float.valueOf(p.get(1).toString());

                                MetricsResult result = new MetricsResult();
                                result.setAppKey(value.getAppKey());
                                result.setAppId(value.getAppId());
                                result.setMetricName(m.getMetricName());
                                Map tagsMap = m.getTags();
                                if (!CollectionUtil.isNullOrEmpty(tagsMap)) {
                                    result.setTags(tagsMap);
                                }
                                result.setTimeStamp(timeStamp);
                                result.setValue(metricValue);
                                out.collect(result);
                            } else {
                                log.error("metrics format is wrong, the format of every point should be timestamp:value  ");
                            }

                        }
                    }

                }
            }
        } catch (Exception e) {
            log.error("Metrics process failed. ", e);
        }
    }
}
